package com.zzl.study.nacos.discovery.reactive;

import com.alibaba.nacos.api.exception.NacosException;
import com.zzl.study.nacos.discovery.NacosServiceMultiDiscovery;
import com.zzl.study.nacos.discovery.ServiceCache;
import lombok.extern.slf4j.Slf4j;
import org.reactivestreams.Publisher;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.discovery.ReactiveDiscoveryClient;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.util.function.Function;

/**
 * Nacos多注册中心响应式服务发现客户端
 *
 * @author: zhangzl
 * @date: 2025/7/16 15:13
 * @version: 1.0
 */
@Slf4j
public class NacosReactiveMultiDiscoveryClient implements ReactiveDiscoveryClient {

    /**
     * 响应式服务发现客户端描述
     */
    public static final String DESCRIPTION = "Spring Cloud Nacos Reactive Multiple Discovery Client";

    /**
     * 失败容错开关
     */
    @Value("${spring.cloud.nacos.multi-discovery.failure-tolerance-enabled:false}")
    private boolean failureToleranceEnabled;

    /**
     * Nacos多注册中心服务发现对象
     */
    private final NacosServiceMultiDiscovery nacosServiceMultiDiscovery;

    public NacosReactiveMultiDiscoveryClient(NacosServiceMultiDiscovery nacosServiceMultiDiscovery) {
        this.nacosServiceMultiDiscovery = nacosServiceMultiDiscovery;
    }

    @Override
    public String description() {
        return DESCRIPTION;
    }

    @Override
    public Flux<ServiceInstance> getInstances(String serviceId) {
        return Mono.justOrEmpty(serviceId).flatMapMany(loadInstancesFromNacos()).subscribeOn(Schedulers.boundedElastic());
    }

    @Override
    public Flux<String> getServices() {
        return Flux.defer(() -> {
            try {
                return Mono.justOrEmpty(this.nacosServiceMultiDiscovery.getServices())
                        .flatMapMany(services -> {
                            ServiceCache.setServiceIds(services);
                            return Flux.fromIterable(services);
                        });
            } catch (Exception e) {
                log.error("get services from nacos server fail,", e);
                return failureToleranceEnabled ? Flux.fromIterable(ServiceCache.getServiceIds()) : Flux.empty();
            }
        }).subscribeOn(Schedulers.boundedElastic());
    }

    /**
     * 从Nacos服务发现获取服务实例列表
     *
     * @return 服务实例列表
     */
    private Function<String, Publisher<ServiceInstance>> loadInstancesFromNacos() {
        return serviceId -> {
            try {
                return Mono.justOrEmpty(this.nacosServiceMultiDiscovery.getInstances(serviceId))
                        .flatMapMany(instances -> {
                            ServiceCache.setInstances(serviceId, instances);
                            return Flux.fromIterable(instances);
                        });
            } catch (NacosException e) {
                log.error("get service instance[{}] from nacos error!", serviceId, e);
                return failureToleranceEnabled ? Flux.fromIterable(ServiceCache.getInstances(serviceId)) : Flux.empty();
            }
        };
    }

}
